/*
 *  RingerFS.cpp
 *  MetaThrift
 *
 *  Created by Ian Pye on 3/16/08.
 *  Copyright 2008 TriceratopsPowered. All rights reserved.
 *
 * FUSE client == interfaces with the MDS and other clients via Thrift
 *
 */

// BSD, stdlib headers
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <iostream>
#include <stdio.h>
#include <string.h>
#include <string>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <sys/time.h>
#include <vector>
#include <map>

// Logger headers
#include <log4cxx/logger.h>
#include <log4cxx/basicconfigurator.h>
#include <log4cxx/helpers/exception.h>
#include <log4cxx/propertyconfigurator.h>

// Thrift headers
#include <protocol/TBinaryProtocol.h>
#include <transport/TSocket.h>
#include <transport/TTransportUtils.h>

// My headers
#include "Ringer.h"
#include "ringer_constants.h"
#include "RingerMetaServer.h"
#include "RingerFS.h"
#include "FileUtil.h"
#include "NetworkUtil.h"
#include "SimHash.h"

// namespi
using namespace std;
using namespace ringer;
using namespace log4cxx;
using namespace log4cxx::helpers;

// general const globals
static const char* RINGER_SEARCH_XATTR = "tag";
static const mode_t RINGER_DIR_CREATE_MODE = 0755;
static const mode_t RINGER_FILE_CREATE_MODE = 0644;

static RingerNetworkArgs *g_net_args; // Global connection info.

// Simhash Stuff
CTags* pTags;
CResults* pResults;

/**
 Update the metadata of the path
 */
static int update_md(const char *path)
{
  RNode rn;
  try{
    ringer_client_get(path, rn);
    RnodeFromPath(path, rn);
    ringer_client_put(rn);
    
    return 0;
  }
  catch (InvalidOperation iop){
    return -1;
  }
}

/**
 Get Attr.
 */
static int
ringer_getattr(const char *path, struct stat *stbuf)
{
  try{
    RNode rn;
    set<rpath_t>::iterator it;
    
    ringer_client_get(path, rn);
    StatFromRNode(rn, stbuf);
    
    LOG4CXX_INFO(g_logger, string("got attr for ") + path);
    for (it=rn.files.begin(); it!=rn.files.end(); it++){
      LOG4CXX_INFO(g_logger, string("has files: ") + *it);    
    }
  } catch (InvalidOperation& eop){
    return -ENOENT;
  } catch (facebook::thrift::TException& ett){
    LOG4CXX_WARN(g_logger, string("Error: ") + ett.what());    
    return -ENOENT;
  }
 	
  return 0;
}

/**
 Create a directory.
 
 @TODO -- Make work with .hosts style path.
 */
static int 
ringer_mkdir (const char *path, mode_t mode){
  
  RNode rn;
  RNode parent;
  string leaf_name;
	int result;
  string parent_str; 
  string local_path;
  LocalizeRingerPath(path, local_path);
  
  LOG4CXX_INFO(g_logger, string("Creating dir ") + path);
  
  // First, get write privelege for the parent.
  ExtractParentDirname(path, parent_str);
  if (!ringer_client_get_write(parent_str.c_str())){  
		LOG4CXX_INFO(g_logger, "denyed getting write priveledge to: " + parent_str);
		return -ENOENT;
	}
  
  // Then put in in the local system.
	result = mkdir (local_path.c_str(), mode);
	
  // Now put the rnode in the metaserver.
  RnodeFromPath(path, rn);
  LOG4CXX_INFO(g_logger, "Putting: " + rn.path);
  ringer_client_put(rn);
  
  // and update it's parent
  ringer_client_get(parent_str.c_str(), parent);
  ExtractFilename(path, leaf_name);
  parent.files.insert(leaf_name);
  LOG4CXX_INFO(g_logger, "Added " + leaf_name + " to the parent " + parent.path + " files");
  ringer_client_put(parent);
  
  // and release the write lock
  ringer_client_end_write(parent_str.c_str());
  
	return result;
}

/** Remove a file */
static int ringer_unlink (const char *path){
  RNode rn;
  string leaf_name;
	int result;
  string local_path;
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  string parent; 
  ExtractParentDirname(path, parent);
  if (!ringer_client_get_write(parent.c_str()) || !ringer_client_get_write(path) ){  
		LOG4CXX_INFO(g_logger, "denyed getting write priveledge on unlink.");
		return -EACCES;
	}
	
  // first remove the rnode in the metaserver.
  ringer_client_remove(path);
  
  // then put in in the local system.
	result = unlink(local_path.c_str());
  if(result < 0){
    ringer_client_end_write(parent.c_str());
    return result;
  }
  
  // and update it's parent
  ringer_client_get(parent.c_str(), rn);
  ExtractFilename(path, leaf_name);
  rn.files.erase(leaf_name);
  LOG4CXX_INFO(g_logger, "Removed " + leaf_name + " from the parent's files");
  ringer_client_put(rn);
  
  // and release the write lock
  ringer_client_end_write(parent.c_str());
  
	return result;
}

/** Remove a directory */
static int ringer_rmdir (const char *path){
  RNode rn;
  string leaf_name;
	int result;
	string local_path;
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  string parent; 
  ExtractParentDirname(path, parent);
  if (!ringer_client_get_write(parent.c_str()) || !ringer_client_get_write(path) ){  
		LOG4CXX_INFO(g_logger, "denyed getting write priveledge on unlink.");
		return -EACCES;
	}
	
  // first remove the rnode in the metaserver.
  ringer_client_remove(path);
  
  // then put in in the local system.
	result = rmdir(local_path.c_str());
  if(result < 0){
    ringer_client_end_write(parent.c_str());
    return result;
  }
  
  // and update it's parent
  ringer_client_get(parent.c_str(), rn);
  ExtractFilename(path, leaf_name);
  rn.files.erase(leaf_name);
  LOG4CXX_INFO(g_logger, "Removed " + leaf_name + " from the parent's files");
  ringer_client_put(rn);
  
  // and release the write lock
  ringer_client_end_write(parent.c_str());
  
	return result;  
}

/** Create a file node
 *
 * This is called for creation of all non-directory, non-symlink
 * nodes.  If the filesystem defines a create() method, then for
 * regular files that will be called instead.
 * -- @TODO -- Make work with .hosts style path.
 */
static int 
ringer_mknod(const char *path, mode_t mode, dev_t dev_type){
	
	int result;
	string local_path;
  LocalizeRingerPath(path, local_path);
	
	LOG4CXX_INFO(g_logger, "creating file " +local_path + " in the local cache");
	
  string parent; 
  ExtractParentDirname(path, parent);
  
	if (!ringer_client_get_write(parent.c_str())){  
		LOG4CXX_INFO(g_logger, "denyed getting write priveledge to: " + parent);
		return -ENOENT;
	}
	
	if (S_ISFIFO(mode))
		result = mkfifo(local_path.c_str(), mode);
	else
		result = mknod(local_path.c_str(), mode, dev_type);
  
	if (result == -1){
		LOG4CXX_WARN(g_logger, "failed to create the file " +local_path 
                 + " in the local cache: " + strerror(errno));
		return -errno;  
	} else {
		LOG4CXX_INFO(g_logger, "created the file " +local_path + " in the local cache");
	}
	
  // now, put the new rnode
  RNode rn;
  string leaf_name;
  RnodeFromPath(path, rn);
  ringer_client_put(rn);
  
  // and update it's parent
  ringer_client_get(parent.c_str(), rn);
  ExtractFilename(path, leaf_name);
  rn.files.insert(leaf_name);
  LOG4CXX_INFO(g_logger, "Added " + leaf_name + " to the parent's files");
  ringer_client_put(rn);
  
  
  // and release the write lock
  ringer_client_end_write(parent.c_str());
  
	return result;
}

/**
 * Create and open a file
 *
 * If the file does not exist, first create it with the specified
 * mode, and then open it.
 *
 * If this method is not implemented or under Linux kernel
 * versions earlier than 2.6.15, the mknod() and open() methods
 * will be called instead.
 *
 * Introduced in version 2.5
 * @TODO -- Make work with .hosts style path
 */

static int
ringer_create (const char *path, mode_t mode, struct fuse_file_info *fi){
  
	int result;
	string local_path;
  LocalizeRingerPath(path, local_path);
	
	string parent; 
  ExtractParentDirname(path, parent);
  
	if (!ringer_client_get_write(parent.c_str())){  
		LOG4CXX_INFO(g_logger, "denyed getting write priveledge to: " + parent);
		return -ENOENT;
	}
	
	result = open(local_path.c_str(), fi->flags | O_CREAT, mode);
	if (result == -1){
		LOG4CXX_WARN(g_logger, "failed to create the file " +local_path 
                 + " in the local cache: " + strerror(errno));
		return -errno;
	} else {
		LOG4CXX_INFO(g_logger, "created the file " +local_path + " in the local cache");
	}
  
	fi->fh = result;
	g_open_files.insert (pair< pair<string, int>, int > 
                       (pair<string, int>(local_path, fi->flags), result));
  
  // now, put the new rnode
  RNode rn;
  string leaf_name;
  RnodeFromPath(path, rn);
  ringer_client_put(rn);
  
  // and update it's parent
  ringer_client_get(parent.c_str(), rn);
  ExtractFilename(path, leaf_name);
  rn.files.insert(leaf_name);
  LOG4CXX_INFO(g_logger, "Added " + leaf_name + " to the parent's files");
  ringer_client_put(rn);
  
  // and release the write lock
  ringer_client_end_write(parent.c_str());
	
	return 0;
}

/**
 Opens a file for access. If the file isn't in the local cache already, downloads. it
 
 if ((fi->flags & O_ACCMODE) != O_RDONLY) { 
 return -EACCES;
 }
 
 */

static int
ringer_open(const char *path, struct fuse_file_info *fi){
	
  string local_path;
  RNode rn;
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  // on open, see if we want to read, read/write or just write
#if !defined(_POSIX_C_SOURCE) || defined(_DARWIN_C_SOURCE)  
  if (fi->flags & (O_WRONLY | O_RDWR | O_CREAT | O_TRUNC | O_EXLOCK))
#else
    if (fi->flags & (O_WRONLY | O_RDWR | O_CREAT | O_TRUNC))
#endif  
    {
      if (!ringer_client_get_write(path)){
        LOG4CXX_INFO(g_logger, string("denyed getting write priveledge to: ") + path);
        return -EACCES;
      }
    } else {
      if (!ringer_client_get_read(path)){
        LOG4CXX_INFO(g_logger, string("denyed getting read priveledge to: ") + path);
        return -EACCES;
      }
    } 
  
  // if the files isn't local, grab the first chunk of it
  if (!IsFileInLocalCacheAndStillCurrent(local_path.c_str(), rn)){     
    LOG4CXX_INFO(g_logger, string("fetching first chunk of ") + path);
    if(!read_network_file(path, ringer::g_ringer_constants.RINGER_BLOCK_SIZE, 0)){
      LOG4CXX_INFO(g_logger, string("error getting first chunk of ") + path);
      return -EACCES;
    }
  }
  
  LOG4CXX_INFO(g_logger, "opening file " +local_path + " from the local cache");
  int lfh = open(local_path.c_str(), fi->flags, RINGER_FILE_CREATE_MODE);
  if (lfh > 0){
    fi->fh = lfh;
    g_open_files.insert (pair< pair<string, int>, int > 
                         (pair<string, int>(local_path, fi->flags), lfh));
  } else {  return -ENOENT; }
  
  LOG4CXX_INFO(g_logger, "successfully opened file " +local_path + "");
  return 0;
}

/**
 Opens and reads a directory
 Note -- this makes 1 rpc for each file in the dir -- need to optimize.
 Also, re-gets read token -- need to cache these.
 @TODO -- no need to aquire a read lock -- this is all just metadata
 */
static int
ringer_readdir(const char *path, void *buf, fuse_fill_dir_t filler,
               off_t offset, struct fuse_file_info *fi) {
  
  RNode rn;
  RNode childrn;
  struct stat *stbuf;
  stbuf = (struct stat *)malloc(sizeof(struct stat));
  
  // get all the files in the dir
  ringer_client_get(path, rn);
  if (!rn.node_type == ringer::RDIRECTORY){
    LOG4CXX_INFO(g_logger, "not a directory: " + rn.path);
    free(stbuf);
    return -ENOENT;
  }
  
  // otherwise, add in all of the files in the local cache
  for (set<rpath_t>::iterator it = rn.files.begin(); it!=rn.files.end(); ++it) {
    rpath_t file_name = path;
    if (file_name.at(file_name.length()-1) != FS_SEPERATOR)
      file_name += FS_SEPERATOR;
    file_name += *it;
    try{
      ringer_client_get(file_name.c_str(), childrn);
      StatFromRNode(childrn, stbuf);
      filler(buf, (*it).c_str(), stbuf, 0);
      LOG4CXX_INFO(g_logger, "Adding file " + *it + " to the list of files returned from read_dir");
    } catch (InvalidOperation ieo){
      LOG4CXX_INFO(g_logger, "File " + file_name + " not in ringer"); 
    }
  }
  
  free(stbuf);
  return 0;
}

/**
 Reads bytes from a file.
 */
static int
ringer_read(const char *path, char *buf, size_t size, off_t offset,
            struct fuse_file_info *fi) {		
  
  string local_path;
  RNode rn;
  
  int lfh;
  size_t read_size;
  
  int myoff_size = size; // read in chunks of ringer size
  if (size >= ringer::g_ringer_constants.RINGER_BLOCK_SIZE && size > 0){
    myoff_size = size % ringer::g_ringer_constants.RINGER_BLOCK_SIZE;
  } else if (size > 0) {
    myoff_size = ringer::g_ringer_constants.RINGER_BLOCK_SIZE % size;
  }
  
  if (!ringer_client_get_read(path)){   // make sure I can read
    LOG4CXX_INFO(g_logger, string("denyed getting read priveledge to: ") + path);
    return -ENOENT;
  }
  
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  lfh = g_open_files.find ( pair<string, int>(local_path, fi->flags) )->second;
  
  if (!IsFileInLocalCacheAndStillCurrent(local_path.c_str(), rn)){ // If not in cache, or our local mtime is earler than the rndoe's mtime, download.
    LOG4CXX_INFO(g_logger, string("Reading chunk from network of ") + path);
    if(!read_network_file(path, size + myoff_size, 0)){
      LOG4CXX_INFO(g_logger, string("error getting chunk of ") + path);
      return -EACCES;  
    }
  }
  
  // try reading the chunk. If it doesn't work for some reason, get the file up to the offset
  LOG4CXX_INFO(g_logger, "reading file " + local_path + " from the local cache");
  read_size = pread(fi->fh, buf, size, offset);
  if (read_size <= 0){
    if(!read_network_file(path, size + myoff_size, 0)){
      LOG4CXX_INFO(g_logger, string("error getting chunk of ") + path);
      return -EACCES;  
    }
    read_size = pread(fi->fh, buf, size, offset);
  }
  return read_size;
}

/**
 Write to a file.
 
 @TODO -- should I do any notifying when I am done?
 
 */
static int 
ringer_write (const char *path, const char *buf, size_t n_bytes, off_t offset, struct fuse_file_info *fi){
  
  string local_path;
  RNode rn;
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  if (!ringer_client_get_write(path)){   // make sure I can write
    LOG4CXX_INFO(g_logger, string("denyed getting write priveledge to: ") + path);
    return -ENOENT;
  }
  
  int myoff_size = n_bytes; // read in chunks of ringer size
  if (n_bytes >= ringer::g_ringer_constants.RINGER_BLOCK_SIZE && n_bytes > 0){
    myoff_size = n_bytes % ringer::g_ringer_constants.RINGER_BLOCK_SIZE;
  } else if (n_bytes > 0) {
    myoff_size = ringer::g_ringer_constants.RINGER_BLOCK_SIZE % n_bytes;
  }
  
  if (!IsFileInLocalCacheAndStillCurrent(local_path.c_str(), rn)){ // If not in cache, or our local mtime is earler than the rndoe's mtime, download.
    LOG4CXX_INFO(g_logger, string("Reading chunk from network of ") + path);
    if(!read_network_file(path, n_bytes + myoff_size, 0)){
      LOG4CXX_INFO(g_logger, string("error getting chunk of ") + path);
      return -EACCES;  
    }
  }
  
  if ( fi->fh ){
    LOG4CXX_INFO(g_logger, "writting file " + local_path + " in the local cache");
    n_bytes = pwrite(fi->fh, buf, n_bytes, offset);
  } else {
    LOG4CXX_INFO(g_logger, "error writting file " + local_path );
    return n_bytes;
  }	
  
  return n_bytes;
}

/**
 Now, here, I do the notifying
 */
static int
ringer_release (const char *path, struct fuse_file_info *fi){
  string local_path;
  RNode rn;
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  // Get the time
  struct timeval tp;
  struct timezone tzp;
  
  rn.atime.time_sec = tp.tv_sec;
  rn.atime.time_nsec = tp.tv_usec / 1000; // Fudge the nano a bit here!
  
  gettimeofday(&tp, &tzp);
  
  // on releaae, see if we want to read, read/write or just write
#if !defined(_POSIX_C_SOURCE) || defined(_DARWIN_C_SOURCE)
  if (fi->flags & (O_WRONLY | O_RDWR | O_CREAT | O_TRUNC | O_EXLOCK))
#else
    if (fi->flags & (O_WRONLY | O_RDWR | O_CREAT | O_TRUNC))
#endif
    {
      // Notify all of the readers that I have modified the file. 
      rn.mtime.time_sec = tp.tv_sec;
      rn.mtime.time_nsec = tp.tv_usec / 1000; // Fudge the nano a bit here!
      
      // Update the simhash value of this file.
      if (ProcessFile(const_cast<char *>(local_path.c_str()), pTags, pResults)){
        for (int j = 0; j < pTags->GetKeyCount(); j++){
          rn.simhash_keys.push_back(pResults->ComputeHashKey(pTags, j, USE_EXT_SIMHASH));
        }
      }
      
      ringer_client_end_write(path);
    } else {
      ringer_client_end_read(path);
    } 
  
  ringer_client_put(rn); // put 'er back.
  
  //@TODO -- make this better
  // Update the rnode values for everything here.
  update_md(path);
  
  if ( fi->fh ){
    LOG4CXX_INFO(g_logger, "releasing file " +local_path + " from the local cache");
    g_open_files.erase(pair<string, int>(local_path, fi->flags));
    return close(fi->fh);
  } else {
    return 0;
  }	
}

/**
 Truncate file = need write access?
 */
static int 
ringer_truncate(const char *path, off_t length){
  
  RNode rn;
  rpath_t local_path;
  ringer_client_get(path, rn);
  LocalizeRingerPath(rn.fuse_path.c_str(), local_path);
  
  if (!ringer_client_get_write(path)){   // make sure I can write
    LOG4CXX_INFO(g_logger, string("denyed getting write priveledge to: ") + path);
    return -ENOENT;
  }
  
  if (IsFileInLocalCache(local_path.c_str())){
    return truncate(local_path.c_str(), length);
  }
  
  return 0;	
}

/**
 No IDEA what I need this for.
 */
static int 
ringer_flush(const char *, struct fuse_file_info *){
  return 0;
}

/** Synchronize file contents
 *
 * If the datasync parameter is non-zero, then only the user data
 * should be flushed, not the meta data.
 *
 * @TODO -- call flush here?
 */
static int 
ringer_fsync (const char *path, int isdatasync, struct fuse_file_info *fi){
  return fsync(fi->fh);
}

/** Set extended attributes */
/** @TODO -- should xposition do anything?
 Also, should there be any locking done here?
 */

static int 
ringer_setxattr (const char *path, const char *name, const char *value, size_t size, int xposition){
  try{
    RNode rn;
    ringer_client_get(path, rn);
    char real_val[size+1];
    strncpy (real_val, value, size);
    real_val[size] = '\0';
    LOG4CXX_INFO(g_logger, (size));
    if (strcmp(name, RINGER_SEARCH_XATTR) == 0){
      LOG4CXX_INFO(g_logger, "Adding tag " + string(real_val));
      rn.keywords.insert(real_val);
    } else {
      rn.xattrs[name] = real_val;
    }
    ringer_client_put(rn);
  } catch (InvalidOperation eop){
    return -ENOENT;
  }
  return 0;
}

/** Get extended attributes */
static int ringer_getxattr (const char *path, const char *name, char *value, size_t size){
  RNode rn;
  map<string,string>::iterator xvalue;
  set<string>::iterator tags;
  int length = 0;
  LOG4CXX_INFO(g_logger, "Getting xattrs for " + string(path));
  try{
    ringer_client_get(path, rn);
    if (strcmp(name, RINGER_SEARCH_XATTR) == 0){
      string tags_str = "";
      for ( tags= rn.keywords.begin() ; tags !=  rn.keywords.end(); tags++ ){
        tags_str += *tags + ", ";
        strncpy(value, tags_str.c_str(), size);
        length = tags_str.length() + 1;
      }
    } else {
      xvalue = rn.xattrs.find(name);
      if(xvalue == rn.xattrs.end()){
        return 0;
      }
      strncpy(value, xvalue->second.c_str(), size);
      length = xvalue->second.length();
    }
  } catch (InvalidOperation eop){
    LOG4CXX_INFO(g_logger, "Caught exception : " + eop.why);
    return -ENOENT;
  }
  
  return length;
}

/** List extended attributes */
static int 
ringer_listxattr (const char *path, char *namebuffer, size_t size){
  string names = "";
  try{
    RNode rn;
    map<string,string>::iterator it;
    ringer_client_get(path, rn);
    names += RINGER_SEARCH_XATTR;
    names += " ";
    for ( it=rn.xattrs.begin() ; it != rn.xattrs.end(); it++ ){
      names += it->first + " ";  
    }
  } catch (InvalidOperation eop){
    return -ENOENT;
  }
  strncpy(namebuffer, names.c_str(), size);
  return names.length()+1;
}

/** Remove extended attributes */
static int 
ringer_removexattr (const char *path, const char *name){
  try{
    RNode rn;
    ringer_client_get(path, rn);
    rn.xattrs.erase(rn.xattrs.find(name));
    ringer_client_put(rn);
  } catch (InvalidOperation eop){
    return -ENOENT;
  }
  return 0;
}


/**
 Pure Metadata operation
 */
static int 
ringer_utimens(const char *path, const struct timespec tv[2]){
  
  RNode rn;
  try{
    ringer_client_get(path, rn);
    
    rn.atime.time_sec = tv[0].tv_sec;
    rn.atime.time_nsec = tv[0].tv_nsec;
    
    rn.mtime.time_sec = tv[1].tv_sec;
    rn.mtime.time_nsec = tv[1].tv_nsec;
    
    rn.ctime.time_sec = tv[2].tv_sec;
    rn.ctime.time_nsec = tv[2].tv_nsec;
    
    ringer_client_put(rn);
    
    return 0;
  }
  catch (InvalidOperation iop){
    return -1;
  }
}

/**
 This is called like constructor for the FS
 */
static void* 
ringer_init(struct fuse_conn_info *conn){
  
  DIR *dp; 
  
  // Set up a simple configuration that logs on the console.
  try{
    PropertyConfigurator::configure(g_net_args->logging_conifg);		
  }
  catch(log4cxx::helpers::Exception &e){
    cerr << "Error: Could not set up the logger: " << endl;
  }
  
  LOG4CXX_INFO(g_logger, "Entering Init.");
  
  // make the cache directory, if it does not exist
  if (((dp  = opendir(g_net_args->local_cache_dir.c_str())) == NULL)){
    mkdir(g_net_args->local_cache_dir.c_str(), RINGER_DIR_CREATE_MODE);
    LOG4CXX_INFO(g_logger, "Createing new local cache dir at " + (g_net_args->local_cache_dir) + ".");
  }
  
  // set this value
  SetRingerLCache(g_net_args->local_cache_dir);
  SetRingeDirCreateMode(RINGER_DIR_CREATE_MODE);
  
  // crank up the thrift client
  ringer_init_network(g_net_args);  
  LOG4CXX_INFO(g_logger, "Thrift client initialized.");
  
  // Set up the SimHash stuff
  pTags = new CTags(const_cast<char *> (g_net_args->simhash_tags_file.c_str()));
	pResults = new CResults(pTags->GetTagCount());
  
  // try putting a rnode for root
  RNode rn;
  RNode kid;
  
  // @TODO -- remove this at some point! 
  //ringer_client_remove(RINGER_ROOT.c_str());
  try{
    ringer_client_get(RINGER_ROOT.c_str(), rn);
    LOG4CXX_INFO(g_logger, "Got root rnode: " + rn.path + "this is a test");
  } catch (InvalidOperation eio){ // didn't find root, add one in.
    RnodeFromPath(RINGER_ROOT.c_str(), rn);
    ringer_client_put(rn);
    LOG4CXX_INFO(g_logger, "Putting new root rnode: " + rn.path);
    
    // also, add in all of the files which the root node has listed as kids
    vector<string> files;
    ListAllSubFiles(RINGER_ROOT.c_str(), files);
    for (int i=0; i<files.size(); i++){
      LOG4CXX_INFO(g_logger, "Adding: " + files.at(i));
      RnodeFromPath(files.at(i).c_str(), kid);
      ringer_client_put(kid);
      LOG4CXX_INFO(g_logger, "Putting new rnode: " + kid.path);
    }
  }
  
  LOG4CXX_INFO(g_logger, "Ending Init.");
  
  return NULL;
}

/**
 A destructor for the FS.
 */
static void
ringer_destroy(void* param){
  LOG4CXX_INFO(g_logger, "Entering Ringer Destroy");
  
  // shut down the network
  ringer_close_network();
  LOG4CXX_INFO(g_logger, "Thrift client stopped.");
  
  LOG4CXX_INFO(g_logger, "Leaving Ringer Destroy");
}

static struct fuse_operations ringer_filesystem_operations = {
ringer_getattr,
0, // readlink
0, // getdir
ringer_mknod, // mknod
ringer_mkdir, // mkdir
ringer_unlink, // unlink
ringer_rmdir, // rmdir
0, // symlink
0, // rename
0, // link
0, // chmod
0, // chown
ringer_truncate, // truncate
0, // utime
ringer_open, // open
ringer_read, // read
ringer_write, // write
0, // statfs
ringer_flush, // flush
ringer_release, // release
ringer_fsync, // sync
ringer_setxattr, // set xattr
ringer_getxattr, // get xattr
ringer_listxattr, // list xattr
ringer_removexattr, // rxattr
0, // opendir
ringer_readdir, // readdir
0, // releasedir
0, // syncdir
ringer_init, // init
ringer_destroy, // destroy
0, // access
ringer_create, // create
0,  // ftruncate
0, // fgetattr
0, //lock
ringer_utimens, //utimens
0 //bmap 
};

int
main(int argc, char **argv) { 
    
  g_net_args = new RingerNetworkArgs();
  g_net_args->mds_port = RINGER_DEFUALT_MDS_PORT;
  g_net_args->self_port = RINGER_DEFUALT_SELF_PORT;
  g_net_args->mds_host = RINGER_DEFAULT_MDS_HOST;
  g_net_args->self_host = RINGER_DEFAULT_SELF_HOST;
  g_net_args->local_cache_dir = RINGER_L_CACHE;
  int fuse_argc = argc;
  char **fuse_argv = argv;
  
  fuse_argc = parse_ringer_args(argc, argv, g_net_args, fuse_argv);
  
  cout << "MDS on " << g_net_args->mds_host << ":" << g_net_args->mds_port << endl;      
  cout << "Self at " << g_net_args->self_host << ":" << g_net_args->self_port << endl;  
  cout << "Local FS Cache at " << g_net_args->local_cache_dir << endl;    
  
  return fuse_main(fuse_argc, fuse_argv, &ringer_filesystem_operations, NULL);
}
